package com.lzwk.app.dwd.ocpx;

import com.lzwk.base.BaseSql;
import com.lzwk.common.ocpx.OCPXConstant;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
 *@Author:yeaser
 *@Date:2022/3/14-17:51
 *Decription:com.lzwk.app.dwd.ocpx
 *@Version:1.8
 */
public class DwdToDoris extends BaseSql {
    public static void main(String[] args) {
        new DwdToDoris().init(4,"ck","dwd_ocpx_pay_order");
    }

    @Override
    protected void run(StreamTableEnvironment tenv) {
        tenv.executeSql("CREATE TABLE source_table (" +
                "   payer_id int," +
                "   product_id int, " +
                "   wk_account_id int, " +
                "   parent_activity_id int, " +
                "   src_prod_id int, " +
                "   org_id int , " +
                "   ks_aid STRING, " +
                "   product_name STRING, " +
                "   skin_id INT, " +
                "   skin_name STRING, " +
                "   source_id INT, " +
                "   source_name STRING, " +
                "   wxapp_id INT, " +
                "   wxapp_nickname STRING, " +
                "   nickname STRING, " +
                "   headimgurl STRING, " +
                "   phone_num STRING, " +
                "   creative_id STRING, " +
                "   ad_group_id STRING, " +
                "   gdt_group_id STRING, " +
                "   fee INT, " +
                "   create_time STRING, " +
                "   pay_time STRING, " +
                "   subscribed INT, " +
                "   id INT, " +
                "   tag_id INT, " +
                "   tag_name STRING, " +
                "   source_platform STRING, " +
                "   refund_status INT, " +
                "   source_platform_order INT " +
                " )WITH (" +
                "   'connector'='kafka', " +
                "   'topic'='" + OCPXConstant.OCPXTESTTOPIC2 + "', " +
                "   'properties.bootstrap.servers'='" + OCPXConstant.TESTKAFKABROKERS + "', " +
                "   'properties.group.id'='dwd_ocpx_pay_order_doris', " +
                "   'scan.startup.mode'='latest-offset', " +
                "   'format'='json'" +
                ")");

        tenv.executeSql("create table dwd_ocpx_pay_order(" +
                "   `payer_id` INT, " +
                "   `product_id` INT, " +
                "   `wk_account_id` INT, " +
                "   `parent_activity_id` INT, " +
                "   `src_prod_id` INT, " +
                "   `org_id` INT, " +
                "   `ks_aid` STRING, " +
                "   `product_name` STRING, " +
                "   `skin_id` INT, " +
                "   `skin_name` STRING, " +
                "   `source_id` INT, " +
                "   `source_name` STRING, " +
                "   `wxapp_id` INT," +
                "   `wxapp_nickname` STRING," +
                "   `nickname` STRING," +
                "   `headimgurl` STRING," +
                "   `phone_num` STRING, " +
                "   `creative_id` STRING, " +
                "   `ad_group_id` STRING, " +
                "   `gdt_group_id` STRING, " +
                "   `fee` INT, " +
                "   `create_time` STRING, " +
                "   `pay_time` STRING, " +
                "   `subscribed` INT, " +
                "   `id` INT, " +
                "   `tag_id` INT, " +
                "   `tag_name` STRING, " +
                "   `source_platform` STRING, " +
                "   `refund_status` INT, " +
                "   `source_paltform_order` INT, " +
                "   PRIMARY KEY (`payer_id`, `product_id`) NOT ENFORCED" +
                ")with(" +
                "   'connector'='doris', " +
                "   'fenodes'='10.2.16.44:8030', " +
                "   'table.identifier'='warehouse.dwd_ocpx_pay_order_test', " +
                "   'username'='flink_cdc'," +
                "   'password'='3H3ekc85eBUGNNj4'," +
                "   'sink.batch.size'='80000', " +
                "   'sink.batch.interval'='3s' " +
                ")");



        tenv.executeSql("  insert into  dwd_ocpx_pay_order " +
                " select " +
                " ifnull(payer_id,-1)  as payer_id, " +
                " ifnull(product_id,-1) as product_id, " +
                " ifnull(wk_account_id,-1) as wk_account_id, " +
                " ifnull(parent_activity_id,-1) as parent_activity_id, " +
                " ifnull(src_prod_id,-1) as src_prod_id, " +
                " ifnull(org_id,-1) as org_id,  " +
                " ifnull(ks_aid,'') as ks_aid, " +
                " ifnull(product_name,'') as product_name, " +
                " ifnull(skin_id,-1) as skin_id, " +
                " ifnull(skin_name,'') as skin_name, " +
                " ifnull(source_id,-1)  as source_id, " +
                " ifnull(source_name,'') as source_name, " +
                " ifnull(wxapp_id,-1) as wxapp_id, " +
                " ifnull(wxapp_nickname,'') as wxapp_nickname,  " +
                " ifnull(nickname,'') as nickname, " +
                " ifnull(headimgurl,'') as headimgurl, " +
                " ifnull(phone_num,'') as phone_num, " +
                " ifnull(creative_id,'') as creative_id, " +
                " ifnull(ad_group_id,'') as ad_group_id, " + // MP消耗与订单关联必-1) 须做限制
                " ifnull( gdt_group_id,'') as gdt_group_id, " + //--广点通消耗表与订-1) 单表关联
                " ifnull(fee,0) as fee, " +
                " ifnull(create_time,'') as create_time, " +
                " ifnull(pay_time,'') as pay_time, " +
                " ifnull(subscribed,0) as subscribed, " +
                " ifnull(id,-1) as id,  " +
                " ifnull(tag_id,100) as tag_id, " +
                " ifnull(tag_name ,'') as tag_name, " +
                " ifnull(source_platform,'')  as source_platform, " +
                " ifnull(refund_status,0) as refund_status, " +
                " source_platform_order  " +
                " from source_table "  );
    }
}
